package com.base.core.producer.broker;

import com.base.api.Message;
import com.base.api.MessageType;
import com.base.core.producer.constant.BrokerMessageConst;
import com.base.core.producer.constant.BrokerMessageStatus;
import com.base.core.producer.entity.BrokerMessage;
import com.base.core.producer.service.MessageStoreService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.apache.commons.lang3.time.DateUtils;
import org.springframework.stereotype.Component;

import java.util.Date;
import java.util.List;

/**
 * 具体发送不同种类型的消息接口
 * 真正的发送不通消息
 */
@Slf4j
@Component
public class RabbitBrokerImpl  implements  RabbitBroker {

    /**
     * 1、rabbittemplate单例,性能会很慢，做池化，提高多生成者发送
     * 2、有的rabbittemplate单例有指定化的需求
     */
//    @Autowired
//    private RabbitTemplate rabbitTemplate;

   private RabbitTemplateConatiner rabbitTemplateConatiner;

   private MessageStoreService messageStoreService;


   public RabbitBrokerImpl(RabbitTemplateConatiner rabbitTemplateConatiner,
                           MessageStoreService messageStoreService){
       this.rabbitTemplateConatiner=rabbitTemplateConatiner;
       this.messageStoreService=messageStoreService;
   }

    @Override
    public void reliantSend(Message message) {
       message.setMessageType(MessageType.RELIANT);
       BrokerMessage bm=messageStoreService.selectByMessageId(message.getMessageId());
       if(bm==null) {
           //1、把数据库的消费发送日志记录好
           Date now = new Date();
           BrokerMessage brokerMessage = new BrokerMessage();
           brokerMessage.setMessageId(message.getMessageId());
           brokerMessage.setStatus(BrokerMessageStatus.SENDING.getCode());
           //tryCount 在最开始发送的时候不需要进行设置
           brokerMessage.setNextRetry(DateUtils.addMinutes(now, BrokerMessageConst.TIMEOUT));
           brokerMessage.setCreateTime(now);
           brokerMessage.setUpdateTime(now);
           brokerMessage.setMessage(message);
           messageStoreService.insert(brokerMessage);
       }
        //2、执行真正的发送消息逻辑
        sendKernel(message);
    }


    /**
     * 迅速发消息
     * @param message
     */
    @Override
    public void rapidSend(Message message) {
            message.setMessageType(MessageType.RAPID);
            sendKernel(message);
    }

    /**
     * 发送消息的核心方法，【发送消息的核心方法，使用异步线程池进行发送消息】
     * 1、亮点一 AsyncBaseQueue线程此的方式发送
     * 2、亮点二 池化的rabbitTemplteConatiner
     *
     * @param message
     */
    private void sendKernel(Message message) {
        AsyncBaseQueue.submit(new Runnable() {
            @Override
            public void run() {
                CorrelationData correlationData=new CorrelationData(String.format("%s#%s#%s",
                        message.getMessageId(),
                        System.currentTimeMillis(), message.getMessageType())
                    );
                String topic=message.getTopic();
                String routingKey=message.getRoutingKey();
                RabbitTemplate rabbitTemplate= rabbitTemplateConatiner.getTemplate(message);
                rabbitTemplate.convertAndSend(topic,routingKey,message,correlationData);
                log.info("#RabbitBrokerImpl.sendKernel# send to rabbitmq,messageId:{} ",message.getMessageId());
            }
        });
     }

    @Override
    public void confirmSend(Message message) {
        message.setMessageType(MessageType.CONFRIM);
        sendKernel(message);
    }

    @Override
    public void sendMessages() {
       List<Message> messages= MessageHolder.clear();
        messages.forEach(message -> {
            MessageHolderAsyncQueue.submit(new Runnable() {
                @Override
                public void run() {
                    CorrelationData correlationData=new CorrelationData(String.format("%s#%s#%s",
                            message.getMessageId(),
                            System.currentTimeMillis(), message.getMessageType())
                    );
                    String topic=message.getTopic();
                    String routingKey=message.getRoutingKey();
                    RabbitTemplate rabbitTemplate= rabbitTemplateConatiner.getTemplate(message);
                    rabbitTemplate.convertAndSend(topic,routingKey,message,correlationData);
                    log.info("#RabbitBrokerImpl.sendMessages# send to rabbitmq,messageId:{} ",message.getMessageId());
                }
            });
        });
    }
}
